package com.le.sarrs.web.rx;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.rx.Promise;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * replace Promise.when
 * Created by yanghongge on 2014/12/29.
 */
public class PromiseWhen<O> {

    Logger logger = LoggerFactory.getLogger(PromiseWhen.class);

    /**
     * return List by promises used thread pool
     * @param tList result list
     * @param timeout sec
     * @return ArrayList<O>
     */
    public ArrayList<O> onSuccess(List<Promise<O>> tList, int timeout) {
        final ReentrantLock lock = new ReentrantLock();
        final Condition pendingCondition = lock.newCondition();
        lock.lock();
        ArrayList<O> list = Lists.newArrayList();
        try {
            long msTimeout = TimeUnit.MILLISECONDS.convert(timeout, TimeUnit.SECONDS);
            long endTime = System.currentTimeMillis() + msTimeout;
            while (System.currentTimeMillis() < endTime && tList.size() > 0) {
                List<Promise<O>> delList = Lists.newArrayList();
                for (Promise<O> promise : tList) {
                    if (promise.isSuccess()) {
                        O o = promise.get();
                        if (null == o) {
                            delList.add(promise);
                            continue;
                        }
                        list.add(o);
                        delList.add(promise);
                    }
                }
                tList.removeAll(delList);
                if (tList.size() > 0)
                    pendingCondition.await(200, TimeUnit.MILLISECONDS);
                else {
                    break;
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            lock.unlock();
        }
        return list;
    }
}
